package cn.doitedu.olap_agg;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.util.Map;

public class Job02_PageAccessTimeLongOlapAggTable {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 构建kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-user-action")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId("gxg")
                .setClientIdPrefix("doitedu-c")
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kfk-dwd");

        DataStream<PageAccess> beanStream = stream.map(json -> JSON.parseObject(json, PageAccess.class));

        SingleOutputStreamOperator<PageAccess> processed = beanStream.keyBy(PageAccess::getSession_id)
                .process(new KeyedProcessFunction<String, PageAccess, PageAccess>() {

                    ValueState<PageAccess> lastPageState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<PageAccess> desc = new ValueStateDescriptor<>("last_page", PageAccess.class);
                        desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.minutes(30)).updateTtlOnReadAndWrite().build());

                        lastPageState = getRuntimeContext().getState(desc);
                    }

                    @Override
                    public void processElement(PageAccess currentEvent, KeyedProcessFunction<String, PageAccess, PageAccess>.Context ctx, Collector<PageAccess> out) throws Exception {

                        String currentEventId = currentEvent.getEvent_id();
                        if (lastPageState.value() == null) {
                            currentEvent.setPage_start_time(currentEvent.getAction_time());
                            lastPageState.update(currentEvent);
                        }


                        // 判断currentEvent是否是page_load事件
                        if (currentEventId.equals("page_load")) {
                            // 将状态中上一个页面的数据替换掉action_time,并输出
                            PageAccess lastPageData = lastPageState.value();
                            lastPageData.setAction_time(currentEvent.getAction_time());
                            out.collect(lastPageData);


                            // 为本条page_load数据，填充页面起始时间，然后替换掉状态中的数据，并输出
                            currentEvent.setPage_start_time(currentEvent.getAction_time());
                            lastPageState.update(currentEvent);
                            out.collect(currentEvent);

                        } else if (currentEventId.equals("wake_up")) {
                            // 填充页面起始时间，然后替换掉状态中的数据，并输出
                            currentEvent.setPage_start_time(currentEvent.getAction_time());
                            lastPageState.update(currentEvent);
                            out.collect(currentEvent);

                        } else {
                            // 别的普通事件，只需要更新掉状态中的  action_time，输出即可
                            PageAccess lastPageData = lastPageState.value();
                            lastPageData.setAction_time(currentEvent.getAction_time());
                            out.collect(lastPageData);
                        }

                        // 如果是app_close事件，则马上清除状态数据
                        if(currentEventId.equals("app_close")){
                            lastPageState.clear();
                        }


                    }
                })
                .filter(bean -> bean.getProperties().get("url") != null);

        // 流转表
        tenv.createTemporaryView("processed", processed,
                Schema.newBuilder()
                        .column("user_id", DataTypes.BIGINT())
                        .column("session_id", DataTypes.STRING())
                        .column("action_time", DataTypes.BIGINT())
                        .column("page_start_time", DataTypes.BIGINT())
                        .column("properties", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .columnByExpression("url", "properties['url']")
                        .column("province", DataTypes.STRING())
                        .column("city", DataTypes.STRING())
                        .column("region", DataTypes.STRING())
                        .columnByExpression("rt", "to_timestamp_ltz(action_time,3)")
                        .watermark("rt", "rt")
                        .build());


        // 创建doris的映射表
        tenv.executeSql(
                        " CREATE TABLE doris_sink  (                                  "+
                        "      dt                 date                                "+
                        "     ,user_id            bigint                              "+
                        "     ,session_id         string                              "+
                        "     ,province           string                              "+
                        "     ,city               string                              "+
                        "     ,region             string                              "+
                        "     ,url                string                              "+
                        "     ,page_start_time    bigint                              "+
                        "     ,page_end_time      bigint                              "+
                        " ) WITH (                                                    "+
                        "    'connector' = 'doris',                                   "+
                        "    'fenodes' = 'doitedu:8030',                              "+
                        "    'table.identifier' = 'dws.page_access_timelong_olap',    "+
                        "    'username' = 'root',                                     "+
                        "    'password' = 'root',                                     "+
                        "    'sink.label-prefix' = 'doris_label-008'                  "+
                        " )                                                           "
        );


        tenv.executeSql(
                        //" INSERT INTO doris_sink                                           "+
                        " SELECT                                                           "+
                        "   to_date(date_format(to_timestamp_ltz(page_start_time,3),'yyyy-MM-dd'))   dt, "+
                        " 	user_id,                                                       "+
                        " 	session_id,                                                    "+
                        " 	province,                                                      "+
                        " 	city,                                                          "+
                        " 	region,                                                        "+
                        " 	url,                                                           "+
                        " 	page_start_time,                                               "+
                        " 	max(action_time) as page_end_time                              "+
                        " FROM TABLE(                                                      "+
                        "     TUMBLE(TABLE processed,DESCRIPTOR(rt),INTERVAL '1' MINUTE)   "+
                        " )                                                                "+
                        " GROUP BY                                                         "+
                        "   window_start,                                                  "+
                        " 	window_end,                                                    "+
                        " 	user_id,                                                       "+
                        " 	session_id,                                                    "+
                        " 	province,                                                      "+
                        " 	city,                                                          "+
                        " 	region,                                                        "+
                        " 	url,                                                           "+
                        " 	page_start_time                                                "
        ).print();


        env.execute();

    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class PageAccess {

        private Long user_id;
        private String session_id;
        private Long action_time;
        private String event_id;
        private Map<String, String> properties;
        private String province;
        private String city;
        private String region;

        private Long page_start_time;

    }

}


